package com.hbase.tableoperation;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.util.Bytes;

public class HbaseTest {

	private static Configuration configuation;
	private static Connection hbaseConnection;
	static{
		configuation=HBaseConfiguration.create();
		configuation.set("hbase.zookeeper.property.clientPort", "2181");
		configuation.set("hbase.zookeeper.quorum", "192.168.51.253");
		configuation.set("hbase.master", "192.168.51.253:60000");
	}
	public static void main(String[] args) {
		// TODO Auto-generated method stub
		init();
		TableName tableName=TableName.valueOf("java_test_table");
		
		//createTeable(tableName);
		//insertData(tableName);
		//dropTable(tableName);
		//deleteRow(tableName, "112233bbbcccc");
		//queryAll(tableName);
		//queryByCondition1(tableName);
		//queryByCondition2(tableName);
		queryByCondition3(tableName);
		
		close();
	}

	//创建一个到HBase的连接
	public static void init(){
		try {
			hbaseConnection = ConnectionFactory.createConnection(configuation);
		} catch (IOException e) {
			e.printStackTrace();
		}
	}
	
	//关闭HBase连接
	public static void close(){
		try {
			hbaseConnection.close();
		} catch (IOException e) {
			e.printStackTrace();
		}
	}
	
	public static void createTeable(TableName tableName){
		System.out.println("start create table .....");
		try {
			Admin hbaseAdmin=hbaseConnection.getAdmin();
			if(hbaseAdmin.tableExists(tableName)){
				//如果存在要创建的表，则先删除，再创建
				hbaseAdmin.disableTable(tableName);
				hbaseAdmin.deleteTable(tableName);
				System.out.println(tableName+" is exist, delete...");
			}
			HTableDescriptor tableDescriptor=new HTableDescriptor(tableName);
			tableDescriptor.addFamily(new HColumnDescriptor("column_family1"));
			tableDescriptor.addFamily(new HColumnDescriptor("column_family2"));
			tableDescriptor.addFamily(new HColumnDescriptor("column_family3"));
			hbaseAdmin.createTable(tableDescriptor);
		} catch (IOException e) {
			e.printStackTrace();
		}
		System.out.println("start create table successfully.");
	}
	
	public static void insertData(TableName tableName){
		System.out.println("start insert data .....");
		try {
			Table table=hbaseConnection.getTable(tableName);
			/*
			 * 一个PUT代表一行数据，再NEW一个PUT表示第二行数据，每行数据一个唯一的rowkey,
			 * 此时rowkey为PUT构造方法中传入的值
			 */
			Put put=new Put("112233bbbcccc".getBytes());
			put.addColumn("column_family1".getBytes(), null, "aaa".getBytes());//本行数据的第一列
			put.addColumn("column_family2".getBytes(), null, "bbb".getBytes());//本行数据的第二列
			put.addColumn("column_family3".getBytes(), null, "ccc".getBytes());//本行数据的第三列

			table.put(put);
			table.close();
		} catch (IOException e) {
			e.printStackTrace();
		}
		System.out.println("start insert data successfully.");
	}
	
	public static void dropTable(TableName tableName){
		System.out.println("drop table .....");
		try {
			Admin admin=hbaseConnection.getAdmin();
			admin.disableTable(tableName);
			admin.deleteTable(tableName);
		} catch (IOException e) {
			e.printStackTrace();
		}
		System.out.println("drop table successfully");
	}
	
	public static void deleteRow(TableName tableName,String rowkey){
		System.out.println("delete row .....");
		try {
			Table table=hbaseConnection.getTable(tableName);
			List<Delete> list=new ArrayList<Delete>();
			Delete dl=new Delete(rowkey.getBytes());
			list.add(dl);
			
			table.delete(list);
		} catch (IOException e) {
			e.printStackTrace();
		}
		System.out.println("delete row successfully");
	}
	
	public static void queryAll(TableName tableName){
		System.out.println("query all data from table ---"+tableName);
		try {
			Table table=hbaseConnection.getTable(tableName);
			ResultScanner rs=table.getScanner(new Scan());
			for(Result r :rs){
				System.out.println("获得到rowkey ："+new String(r.getRow()));
				for(KeyValue keyValue:r.raw()){
					System.out.println("列："+new String(keyValue.getFamily())
							+"=====值："+new String(keyValue.getValue()));
				}
			}
		} catch (IOException e) {
			e.printStackTrace();
		}
	}
	
	public static void queryByCondition1(TableName tableName){
		
		try {
			Table table=hbaseConnection.getTable(tableName);
			Get get=new Get("112233bbbcccc".getBytes());
			Result r=table.get(get);
			
			System.out.println("获得到rowkey ："+new String(r.getRow()));
			for(KeyValue keyValue:r.raw()){
				System.out.println("列："+new String(keyValue.getFamily())
						+"=====值："+new String(keyValue.getValue()));
			}
			
		} catch (IOException e) {
			e.printStackTrace();
		}
	}
	
	public static void queryByCondition2(TableName tableName){
		
		Filter filter=new SingleColumnValueFilter(
				Bytes.toBytes("column_family1"),null,CompareOp.EQUAL,
				Bytes.toBytes("aaa"));
		
		Scan scan=new Scan();
		scan.setFilter(filter);
		try {
			Table table=hbaseConnection.getTable(tableName);
			ResultScanner rs=table.getScanner(scan);
			for(Result r :rs){
				System.out.println("获得到rowkey ："+new String(r.getRow()));
				for(KeyValue keyValue:r.raw()){
					System.out.println("列："+new String(keyValue.getFamily())
							+"=====值："+new String(keyValue.getValue()));
				}
			}
			rs.close();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
	
	public static void queryByCondition3(TableName tableName){
		
		List<Filter> filters=new ArrayList<Filter>();
		Filter filter1=new SingleColumnValueFilter(Bytes.toBytes("column_family1"),
				null,CompareOp.EQUAL,Bytes.toBytes("aaa"));
		filters.add(filter1);
		
		Filter filter2=new SingleColumnValueFilter(Bytes.toBytes("column_family2"),
				null,CompareOp.EQUAL,Bytes.toBytes("bbb"));
		filters.add(filter2);
		
		Filter filter3=new SingleColumnValueFilter(Bytes.toBytes("column_family3"),
				null,CompareOp.EQUAL,Bytes.toBytes("ccc"));
		filters.add(filter3);
		
		FilterList filterList=new FilterList(filters);
		Scan scan=new Scan();
		scan.setFilter(filterList);
		try {
			Table table=hbaseConnection.getTable(tableName);
			ResultScanner rs=table.getScanner(scan);
			for(Result r :rs){
				System.out.println("获得到rowkey ："+new String(r.getRow()));
				for(KeyValue keyValue:r.raw()){
					System.out.println("列："+new String(keyValue.getFamily())
							+"=====值："+new String(keyValue.getValue()));
				}
			}
			rs.close();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		
	}
}
